package udf.matching;

import java.io.IOException;

import java.util.ArrayList;
import java.util.Hashtable;
import java.util.Iterator;

import org.apache.pig.EvalFunc;
import org.apache.pig.data.BagFactory;
import org.apache.pig.data.DataBag;
import org.apache.pig.data.Tuple;
import org.apache.pig.data.TupleFactory;

/**
 * 
 * Receives a input tuple with the following fields
 * 
 * 1. Tuple with tags of size n 2. Bag with the tags to perform the
 * intersection, typically (something, tag) 3. Pos of the target tag in 2.Bag 4.
 * Position to place the matched tag in 1.Tuple
 * 
 * 
 * @author sergio
 * 
 */
public class JoinUsingBag extends EvalFunc<DataBag> {

	private boolean initialized = false;
	private int bag_entry_size = -1;
	private BagFactory bagFactory = BagFactory.getInstance();

	// static ArrayList<String> mTables = new ArrayList<String>();

	// static HashSet<String> seeds = new HashSet<String>();
	private Hashtable<String, String[]> seeds = new Hashtable<String, String[]>();

	// static private final String domain_path =
	// "/user/duartes/helpers/domains.txt";
/**
 * Tuple has the following fields
 * 
 * 
 * 0  -   tuple with fields to be intersected
 * 
 * 1  -  dictionary to do the look up
 * 
 * 2  -  position of tag in dictionary of (1)
 * 
 * 3  -  resorting of tuples in , 3 spicify the position in 1
 * 
 * 
 */
	public void init(Tuple tuple) throws IOException {

		DataBag bag = (DataBag) tuple.get(1);

		String tag_pos_ = (String) tuple.get(2);
		int tag_pos = Integer.valueOf(tag_pos_);

		Iterator<Tuple> iterator = bag.iterator();

		while (iterator.hasNext()) {

			Tuple tup = iterator.next();
			String temp[] = new String[tup.size()];
			for (int i = 0; i < tup.size(); i++) {
				temp[i] = (String) tup.get(i);
			}

			bag_entry_size = temp.length;

			String tag = (String) tup.get(tag_pos);
			// System.out.println("Adding: " + tag + "\t" + temp[0]);
			seeds.put(tag, temp);

		}

		initialized = true;
	}

	public DataBag exec(Tuple input) throws IOException {
			if (!initialized)
				init(input);

			Tuple tags = (Tuple) input.get(0);
			String pos_ = (String) input.get(3);
			
			
			int debug =0;
			
			if(input.size()>4){
				debug = (Integer)input.get(4);
				
			}
			int pos = Integer.valueOf(pos_);
			ArrayList<Integer> found = new ArrayList<Integer>();
			ArrayList<String[]> lines = new ArrayList<String[]>();

		
			if(debug==1)
			System.out.println("matching:");
			for (int i = 0; i < tags.size(); i++) {

				String tag = (String) tags.get(i);
				if(debug==1)
				System.out.println("\t"+ tag);
				if (seeds.containsKey(tag)) {
					found.add(i);
					lines.add(seeds.get(tag));
					if(debug==1)
					System.out.println("\t"+ "found position:" + i);
				//	System.out.println("Matching " + tag);
				}

			}
			
			DataBag output = bagFactory.newDefaultBag();

			if (found.size() > 0) {
			//	System.out.println("Found size: "+ found);
				for (int k = 0; k < found.size(); k++) {
					
					int index = found.get(k);
					
					//Tuple t = TupleFactory.getInstance().newTuple(tags.size() + bag_entry_size);
					Tuple t = TupleFactory.getInstance().newTuple();
					String temporal = (String) tags.get(pos);
					String target = (String) tags.get(index);
					
					if(debug==1)
					System.out.println("pos: "+ "\t" + target + "\t" + tags.size()  );
					for (int i = 0; i < tags.size(); i++) {
						

						if (i == pos) {
							//t.set(i, target);
							t.append(target);

						} else if (i == index) {
							//t.set(i, temporal);
							t.append(temporal);

						} else {
						//	t.set(i, tags.get(i));
							t.append(tags.get(i));
						}
					}
					
					String line[] = lines.get(k);
				
					
					for (int j = 0; j <line.length; j++) {
				
						t.append(line[j]);
						//t.set(i, line[j]);
					
					}
					
					if(debug==1)
					System.out.println("ADDINNG THE FOLLOWING TUPLE:");
					
					
					for (int j = 0; j <t.size(); j++) {
						if(debug==1)
						System.out.println(j + "\t" + t.get(j));
						//t.set(i, line[j]);
					
					}
					t.append(index);
					
					output.add(t);
				}

			
			return output;

		}
			
			
			
			return null;

	}
}
